package mq

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"task_queue/internal/models"
	"time"

	"gorm.io/gorm"
)

// 定义错误类型
var (
	ErrTopicNotFound = errors.New("主题不存在")
	ErrTopicClosed   = errors.New("主题已关闭")
)

// ChannelQueuePlugin 实现基于 Go channel 的内存队列
type ChannelQueuePlugin struct {
	topics     map[string]chan string
	mu         sync.RWMutex
	closed     bool
	bufSize    int
	startTime  int64
	queueStats map[string]*QueueStats
	statsMutex sync.RWMutex
	wg         sync.WaitGroup // 用于管理 goroutine 生命周期
	db         *gorm.DB       // 添加数据库连接
}

// 确保 ChannelQueuePlugin 实现了 MQPlugin 接口
var _ MQPlugin = (*ChannelQueuePlugin)(nil)

// newChannelQueuePlugin 创建一个新的 ChannelQueuePlugin 实例
func newChannelQueuePlugin(config map[string]interface{}) (MQPlugin, error) {
	// 从配置中获取缓冲区大小，默认为 1000
	bufferSize := 1000
	if size, ok := config["buffer_size"].(int); ok && size > 0 {
		bufferSize = size
	}

	// 获取数据库连接
	var db *gorm.DB
	if dbInterface, ok := config["db"]; ok {
		if dbConn, ok := dbInterface.(*gorm.DB); ok {
			db = dbConn
		}
	}

	cq := &ChannelQueuePlugin{
		topics:     make(map[string]chan string),
		closed:     false,
		bufSize:    bufferSize,
		startTime:  time.Now().Unix(),
		queueStats: make(map[string]*QueueStats), // 初始化 queueStats 映射
		wg:         sync.WaitGroup{},
		db:         db, // 设置数据库连接
	}

	// 预创建默认主题
	if defaultTopic, ok := config["default_topic"].(string); ok && defaultTopic != "" {
		cq.topics[defaultTopic] = make(chan string, bufferSize)
		// 为默认主题创建统计信息
		cq.queueStats[defaultTopic] = &QueueStats{
			TopicName: defaultTopic,
		}
	}

	return cq, nil
}

// Type 返回消息队列类型
func (cq *ChannelQueuePlugin) Type() MQType {
	return ChannelQueue
}

// Publish 发布消息到指定主题
func (cq *ChannelQueuePlugin) Publish(ctx context.Context, topic string, message string) error {
	// 检查队列是否已关闭
	cq.mu.RLock()
	if cq.closed {
		cq.mu.RUnlock()
		return ErrTopicClosed
	}

	// 获取或创建主题
	ch, ok := cq.topics[topic]
	fmt.Println("所有的topics", cq.topics)
	if !ok {
		// 如果主题不存在，创建一个新的
		cq.mu.RUnlock()
		cq.mu.Lock()
		// 再次检查，避免并发创建
		if ch, ok = cq.topics[topic]; !ok {
			ch = make(chan string, cq.bufSize)
			cq.topics[topic] = ch

			// 创建主题的同时初始化统计信息
			cq.statsMutex.Lock()
			cq.queueStats[topic] = &QueueStats{
				TopicName: topic,
			}
			fmt.Println("创建topic", topic)
			cq.statsMutex.Unlock()
		}
		cq.mu.Unlock()
	} else {
		cq.mu.RUnlock()
	}
	fmt.Println("所有的topics", cq.topics)

	// 使用 select 实现带超时的发送
	select {
	case ch <- message:
		// 记录消息历史（如果有数据库连接）
		if cq.db != nil {
			messageHistory := models.MessageHistory{
				Topic:     topic,
				MQType:    ChannelQueue,
				Message:   message,
				Status:    models.MessageStatusPending,
				CreatedAt: time.Now(),
			}

			// 异步保存，避免阻塞发布过程
			go func(history models.MessageHistory) {
				if err := cq.db.Create(&history).Error; err != nil {
					fmt.Printf("记录消息历史失败: %v\n", err)
				}
			}(messageHistory)
		}
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// Subscribe 订阅指定主题的消息
func (cq *ChannelQueuePlugin) Subscribe(ctx context.Context, topic string, handler func(message string) error) error {
	// 检查队列是否已关闭
	cq.mu.RLock()
	if cq.closed {
		cq.mu.RUnlock()
		return ErrTopicClosed
	}

	// 获取或创建主题
	ch, ok := cq.topics[topic]
	if !ok {
		// 如果主题不存在，创建一个新的
		cq.mu.RUnlock()
		cq.mu.Lock()
		// 再次检查，避免并发创建
		if ch, ok = cq.topics[topic]; !ok {
			ch = make(chan string, cq.bufSize)
			cq.topics[topic] = ch

			// 创建主题的同时初始化统计信息
			cq.statsMutex.Lock()
			cq.queueStats[topic] = &QueueStats{
				TopicName: topic,
			}
			cq.statsMutex.Unlock()
		}
		cq.mu.Unlock()
	} else {
		cq.mu.RUnlock()
	}

	// 更新统计信息
	cq.statsMutex.Lock()
	stats, ok := cq.queueStats[topic]
	if !ok {
		stats = &QueueStats{
			TopicName: topic,
		}
		cq.queueStats[topic] = stats
	}
	cq.statsMutex.Unlock()

	// 从通道读取消息并处理
	for {
		select {
		case <-ctx.Done():
			// 上下文取消，不应将未处理的消息标记为失败
			return ctx.Err()
		case msg, ok := <-ch:
			if !ok {
				// 通道已关闭
				return ErrTopicClosed
			}

			// 处理消息
			err := handler(msg)

			// 更新统计信息
			cq.statsMutex.Lock()
			if err == nil {
				stats.ConsumedMessages++
				stats.LastConsumedAt = time.Now().Unix()

				// 更新消息历史为已消费（如果有数据库连接）
				if cq.db != nil {
					now := time.Now()
					// 查找并更新消息状态
					go func(message string, consumedTime time.Time) {
						// 查找最近的一条待处理消息
						var history models.MessageHistory
						if err := cq.db.Where("topic = ? AND mq_type = ? AND message = ? AND status = ?",
							topic, ChannelQueue, message, models.MessageStatusPending).
							Order("created_at DESC").
							First(&history).Error; err == nil {

							// 更新为已消费状态
							history.Status = models.MessageStatusConsumed
							history.ConsumedAt = &consumedTime
							cq.db.Save(&history)
						}
					}(msg, now)
				}
			} else {
				// 只有当处理函数返回错误时，才将消息标记为失败
				// 如果是因为达到消费数量限制而返回的错误，不应增加失败计数
				if !strings.Contains(err.Error(), "已达到指定消费数量") {
					stats.FailedMessages++

					// 更新消息历史为失败（如果有数据库连接）
					if cq.db != nil {
						// 查找并更新消息状态
						go func(message string, errorMsg string) {
							// 查找最近的一条待处理消息
							var history models.MessageHistory
							if err := cq.db.Where("topic = ? AND mq_type = ? AND message = ? AND status = ?",
								topic, ChannelQueue, message, models.MessageStatusPending).
								Order("created_at DESC").
								First(&history).Error; err == nil {

								// 更新为失败状态
								history.Status = models.MessageStatusFailed
								history.ErrorMessage = errorMsg
								cq.db.Save(&history)
							}
						}(msg, err.Error())
					}
				} else {
					// 如果是因为达到消费数量限制，将消息放回队列
					go func(message string) {
						// 使用新的上下文，避免被取消
						newCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
						defer cancel()
						cq.Publish(newCtx, topic, message)
					}(msg)
				}
			}
			cq.statsMutex.Unlock()

			// 如果处理函数返回错误，停止订阅
			if err != nil {
				return err
			}
		}
	}
}

// Close 关闭所有主题和处理程序
func (cq *ChannelQueuePlugin) Close() error {
	cq.mu.Lock()
	defer cq.mu.Unlock()

	if cq.closed {
		return nil
	}

	// 标记为已关闭
	cq.closed = true

	// 关闭所有主题
	for topic, ch := range cq.topics {
		close(ch)
		delete(cq.topics, topic)
	}

	// 等待所有 goroutine 结束
	cq.wg.Wait()

	return nil
}

// GetStatus 获取Channel队列的状态
func (cq *ChannelQueuePlugin) GetStatus() MQStatus {
	cq.mu.RLock()
	defer cq.mu.RUnlock()

	return MQStatus{
		IsConnected:    !cq.closed,
		ConnectionInfo: "内存队列",
		StartTime:      cq.startTime,
		ErrorCount:     0, // 内存队列不会有连接错误
		LastError:      "",
	}
}

// GetQueueStats 获取指定主题的队列统计信息
func (cq *ChannelQueuePlugin) GetQueueStats(topic string) (QueueStats, error) {
	cq.statsMutex.RLock()
	defer cq.statsMutex.RUnlock()

	// 检查队列是否存在
	stats, exists := cq.queueStats[topic]
	if !exists {
		// 如果统计信息不存在但主题存在，则创建统计信息
		cq.mu.RLock()
		_, topicExists := cq.topics[topic]
		cq.mu.RUnlock()

		if topicExists {
			// 创建新的统计信息
			cq.statsMutex.RUnlock()
			cq.statsMutex.Lock()
			stats = &QueueStats{
				TopicName: topic,
			}
			cq.queueStats[topic] = stats
			cq.statsMutex.Unlock()
			cq.statsMutex.RLock()

			// 获取当前队列长度
			cq.mu.RLock()
			ch := cq.topics[topic]
			pendingCount := len(ch)
			cq.mu.RUnlock()

			// 更新统计信息
			stats.PendingMessages = int64(pendingCount)

			return *stats, nil
		}

		return QueueStats{}, fmt.Errorf("主题 %s 不存在或未被监控", topic)
	}

	// 获取当前队列长度
	cq.mu.RLock()
	ch, exists := cq.topics[topic]
	pendingCount := 0
	if exists {
		pendingCount = len(ch)
	}
	cq.mu.RUnlock()

	// 更新统计信息
	stats.PendingMessages = int64(pendingCount)

	return *stats, nil
}

// ListQueueStats 获取所有主题的队列统计信息
func (cq *ChannelQueuePlugin) ListQueueStats() []QueueStats {
	cq.statsMutex.RLock()
	defer cq.statsMutex.RUnlock()

	result := make([]QueueStats, 0, len(cq.queueStats))

	// 更新所有队列的待处理消息数
	cq.mu.RLock()
	for topic, stats := range cq.queueStats {
		if ch, exists := cq.topics[topic]; exists {
			stats.PendingMessages = int64(len(ch))
		}
		result = append(result, *stats)
	}
	cq.mu.RUnlock()

	return result
}

// GetPendingMessages 获取待处理的消息
func (cq *ChannelQueuePlugin) GetPendingMessages(topic string, page, pageSize int) ([]string, error) {
	// 检查主题是否存在
	cq.mu.RLock()
	ch, exists := cq.topics[topic]
	if !exists {
		cq.mu.RUnlock()
		return nil, fmt.Errorf("主题 %s 不存在", topic)
	}

	// 计算分页参数
	if page < 1 {
		page = 1
	}
	if pageSize < 1 {
		pageSize = 10
	}

	// 获取通道中的所有消息（非破坏性读取）
	var allMessages []string
	pendingCount := len(ch)

	// 创建一个临时切片来存储通道中的所有消息
	tempMessages := make([]string, 0, pendingCount)

	// 非破坏性地读取通道中的消息
	for i := 0; i < pendingCount; i++ {
		select {
		case msg := <-ch:
			tempMessages = append(tempMessages, msg)
			// 将消息放回通道
			go func(message string) {
				cq.mu.RLock()
				defer cq.mu.RUnlock()
				if !cq.closed {
					select {
					case ch <- message:
					default:
						// 如果通道已满，记录错误但不阻塞
						fmt.Printf("警告：无法将消息放回主题 %s 的通道，通道可能已满\n", topic)
					}
				}
			}(msg)
		default:
			// 如果通道为空，跳出循环
			break
		}
	}
	cq.mu.RUnlock()

	allMessages = tempMessages

	// 计算分页结果
	startIndex := (page - 1) * pageSize
	endIndex := startIndex + pageSize

	// 检查索引是否越界
	if startIndex >= len(allMessages) {
		return []string{}, nil
	}
	if endIndex > len(allMessages) {
		endIndex = len(allMessages)
	}

	// 返回分页结果
	return allMessages[startIndex:endIndex], nil
}

// GetConsumedMessages 获取已消费的消息
func (cq *ChannelQueuePlugin) GetConsumedMessages(topic string, page, pageSize int) ([]MessageRecord, error) {
	// 检查主题是否存在
	cq.statsMutex.RLock()
	_, exists := cq.queueStats[topic]
	if !exists {
		cq.statsMutex.RUnlock()
		return nil, fmt.Errorf("主题 %s 不存在或未被监控", topic)
	}
	cq.statsMutex.RUnlock()

	// 如果没有数据库连接，返回空列表
	if cq.db == nil {
		return []MessageRecord{}, nil
	}

	// 计算分页参数
	if page < 1 {
		page = 1
	}
	if pageSize < 1 {
		pageSize = 10
	}
	offset := (page - 1) * pageSize

	// 从数据库查询已消费的消息
	var histories []models.MessageHistory
	if err := cq.db.Where("topic = ? AND mq_type = ? AND status = ?",
		topic, ChannelQueue, models.MessageStatusConsumed).
		Order("consumed_at DESC").
		Offset(offset).
		Limit(pageSize).
		Find(&histories).Error; err != nil {
		return nil, err
	}

	// 转换为MessageRecord格式
	records := make([]MessageRecord, len(histories))
	for i, history := range histories {
		var timestamp int64
		if history.ConsumedAt != nil {
			timestamp = history.ConsumedAt.Unix()
		} else {
			timestamp = history.CreatedAt.Unix()
		}

		records[i] = MessageRecord{
			Message:   history.Message,
			Timestamp: timestamp,
		}
	}

	return records, nil
}

// GetFailedMessages 获取失败的消息
func (cq *ChannelQueuePlugin) GetFailedMessages(topic string, page, pageSize int) ([]MessageRecord, error) {
	// 检查主题是否存在
	cq.statsMutex.RLock()
	_, exists := cq.queueStats[topic]
	if !exists {
		cq.statsMutex.RUnlock()
		return nil, fmt.Errorf("主题 %s 不存在或未被监控", topic)
	}
	cq.statsMutex.RUnlock()

	// 如果没有数据库连接，返回空列表
	if cq.db == nil {
		return []MessageRecord{}, nil
	}

	// 计算分页参数
	if page < 1 {
		page = 1
	}
	if pageSize < 1 {
		pageSize = 10
	}
	offset := (page - 1) * pageSize

	// 从数据库查询失败的消息
	var histories []models.MessageHistory
	if err := cq.db.Where("topic = ? AND mq_type = ? AND status = ?",
		topic, ChannelQueue, models.MessageStatusFailed).
		Order("created_at DESC").
		Offset(offset).
		Limit(pageSize).
		Find(&histories).Error; err != nil {
		return nil, err
	}

	// 转换为MessageRecord格式
	records := make([]MessageRecord, len(histories))
	for i, history := range histories {
		records[i] = MessageRecord{
			Message:   history.Message,
			Timestamp: history.CreatedAt.Unix(),
			Error:     history.ErrorMessage,
		}
	}

	return records, nil
}

// CloseQueue 关闭指定主题的队列
func (cq *ChannelQueuePlugin) CloseQueue(topic string) error {
	cq.mu.Lock()
	defer cq.mu.Unlock()

	// 检查主题是否存在
	ch, exists := cq.topics[topic]
	if !exists {
		return fmt.Errorf("主题 %s 不存在", topic)
	}

	// 关闭通道
	close(ch)

	// 从映射中移除
	delete(cq.topics, topic)

	// 从统计信息中移除
	cq.statsMutex.Lock()
	delete(cq.queueStats, topic)
	cq.statsMutex.Unlock()

	return nil
}

// ListTopics 获取所有主题列表
func (cq *ChannelQueuePlugin) ListTopics() []string {
	cq.mu.RLock()
	defer cq.mu.RUnlock()

	topics := make([]string, 0, len(cq.topics))
	for topic := range cq.topics {
		topics = append(topics, topic)
	}

	return topics
}
